package com.example.mysample.bloom;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.BitFieldSubCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Objects;

@Service
public class RedisBloomFilter {
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private BloomFilterHelper bloomFilterHelper;
    @Value("${spring.redis.type:}")
    private String redisType;

    private static final String CLUSTER = "cluster";

    /**
     * 根据给定的布隆过滤器添加值
     */
    public <T> void addElement(String key, T value) {
        if (Objects.equals(CLUSTER, redisType)) {
            return;
        }
        int[] offset = bloomFilterHelper.murmurHashOffset(value);
        /*redisTemplate.executePipelined((RedisCallback) connection -> {
            for (int i : offset) {
                connection.setBit(redisTemplate.getStringSerializer().serialize(key), i, true);
            }
            return null;
        });*/

        BitFieldSubCommands commands = BitFieldSubCommands.create();
        for (int i : offset) {
            commands = commands.set(BitFieldSubCommands.BitFieldType.unsigned(1))
                    .valueAt(i)
                    .to(1);
        }
        redisTemplate.opsForValue().bitField(key, commands);
    }

    /**
     * 根据给定的布隆过滤器判断值是否存在
     */
    public <T> boolean containsElement(String key, T value) {
        if (Objects.equals(CLUSTER, redisType)) {
            return false;
        }
        int[] offset = bloomFilterHelper.murmurHashOffset(value);

        /*List<Object> list = redisTemplate.executePipelined((RedisCallback) connection -> {
            for (int i : offset) {
                connection.getBit(redisTemplate.getStringSerializer().serialize(key), i);
            }
            return null;
        });
        return !list.contains(false);*/


        BitFieldSubCommands commands = BitFieldSubCommands.create();
        for (int i : offset) {
            commands = commands.get(BitFieldSubCommands.BitFieldType.unsigned(1))
                    .valueAt(i);
        }
        List<Long> result = redisTemplate.opsForValue().bitField(key, commands);
        return !result.contains(0L);
    }

}
